package com.small.nacos.config.server.service;


import com.alibaba.nacos.common.http.Callback;
import com.alibaba.nacos.common.http.client.NacosAsyncRestTemplate;
import com.alibaba.nacos.common.http.param.Header;
import com.alibaba.nacos.common.http.param.Query;
import com.alibaba.nacos.common.model.RestResult;
import small.common.http.util.ContextPathUtil;
import small.common.core.util.StringUtil;
import com.small.nacos.api.common.ApiCommonConstants;
import com.small.nacos.api.common.ApiConstants;
import com.small.nacos.api.config.common.ConfigConstants;
import com.small.nacos.common.notify.Event;
import com.small.nacos.common.notify.NotifyCenter;
import com.small.nacos.common.notify.listener.Subscriber;
import com.small.nacos.config.server.model.event.ConfigDataChangeEvent;
import com.small.nacos.config.server.service.notify.ConfigHttpClientManager;
import com.small.nacos.config.server.service.notify.NotifyTask;
import com.small.nacos.config.server.utils.ConfigExecutor;
import com.small.nacos.core.cluster.Member;
import com.small.nacos.core.cluster.ServerMemberManager;
import com.small.nacos.sys.env.EnvUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.TimeUnit;

@Service
public class AsyncNotifyService {

    protected static final Logger LOGGER = LoggerFactory.getLogger(AsyncNotifyService.class);

    /**
     * 最小重试间隔
     */
    private static final int MIN_RETRY_INTERVAL = 500;

    private static final int INCREASE_STEPS = 1000;
    /**
     * 最大重试次数
     */
    private static final int MAX_COUNT = 6;

    protected ServerMemberManager memberManager;

    private final NacosAsyncRestTemplate nacosAsyncRestTemplate = ConfigHttpClientManager.getNacosAsyncRestTemplate();

    public AsyncNotifyService(ServerMemberManager serverMemberManager) {

        this.memberManager = serverMemberManager;

        // 注册事件发布者
        NotifyCenter.registerToPublisher(ConfigDataChangeEvent.class, NotifyCenter.ringBufferSize);

        NotifyCenter.registerSubscriber(new Subscriber() {
            @Override
            public void onEvent(Event event) {
                if (event instanceof ConfigDataChangeEvent) {
                    ConfigDataChangeEvent evt = (ConfigDataChangeEvent) event;
                    long dumpTs = evt.getLastModifiedTs();
                    String dataId = evt.getDataId();
                    String group = evt.getGroup();
                    String tenant = evt.getTenant();
                    Collection<Member> ipList = serverMemberManager.allMembers();
                    Queue<NotifySingleTask> queue = new LinkedList<NotifySingleTask>();
                    for (Member member : ipList) {
                        queue.add(new NotifySingleTask(dataId, group, tenant, dumpTs, member.getAddress()));
                    }

                    ConfigExecutor.executeAsyncNotify(new AsyncTask(nacosAsyncRestTemplate, queue));
                }
            }

            @Override
            public Class<? extends Event> subscribeType() {
                return ConfigDataChangeEvent.class;
            }
        });
    }

    static class NotifySingleTask extends NotifyTask {

        public String target;

        public String url;

        public Map<String,String> paramValues;

        public NotifySingleTask(String dataId, String group, String tenant, long lastModified, String target) {
            super(dataId, group, tenant, lastModified);
            this.target = target;
            this.url = ContextPathUtil.normalizeContextUrl(target,EnvUtil.getContextPath(),ApiConstants.COMMUNICATION_CONTROLLER_PATH,ApiConstants.DATA_CHANGE);
            Map<String, String> paramValues = new HashMap();
            paramValues.put(ConfigConstants.DATA_ID,dataId);
            paramValues.put(ConfigConstants.GROUP,group);
            if (StringUtil.isNotBlank(tenant)) {
                paramValues.put(ConfigConstants.TENANT,tenant);
            }
            this.paramValues=paramValues;
        }

        public String getTargetIP() {
            return target;
        }
    }

    class AsyncTask implements Runnable {

        private Queue<NotifySingleTask> queue;

        private NacosAsyncRestTemplate restTemplate;

        public AsyncTask(NacosAsyncRestTemplate restTemplate, Queue<NotifySingleTask> queue) {
            this.restTemplate = restTemplate;
            this.queue = queue;
        }

        @Override
        public void run() {
            executeAsyncInvoke();
        }

        private void executeAsyncInvoke() {
            while (!queue.isEmpty()) {
                NotifySingleTask task = queue.poll();
                String targetIp = task.getTargetIP();
                if (memberManager.hasMember(targetIp)) {
                    if (!memberManager.isHealth(targetIp)) {
                        //如果当前服务不健康 延时推送任务
                        asyncTaskExecute(task);
                    } else {
                        Header header = Header.newInstance();
                        header.addParam(ApiCommonConstants.NOTIFY_HEADER_LAST_MODIFIED, String.valueOf(task.getLastModified()));

                        Query query = Query.newInstance().initParams(task.paramValues);
                        restTemplate.get(task.url, header, query, String.class, new AsyncNotifyCallBack(task));
                    }
                }
            }
        }
    }

    /**
     * 延时任务执行
     *
     * @param task
     */
    private void asyncTaskExecute(NotifySingleTask task) {
        int delay = getDelayTime(task);
        Queue<NotifySingleTask> queue = new LinkedList();
        queue.add(task);
        AsyncTask asyncTask = new AsyncTask(nacosAsyncRestTemplate, queue);
        ConfigExecutor.scheduleAsyncNotify(asyncTask, delay, TimeUnit.MILLISECONDS);
    }

    /**
     * 获取延时推送任务时间
     *
     * @param task
     * @return
     */
    private static int getDelayTime(NotifySingleTask task) {
        int failCount = task.getFailCount();
        int delay = MIN_RETRY_INTERVAL + failCount * failCount * INCREASE_STEPS;
        if (failCount <= MAX_COUNT) {
            task.setFailCount(failCount + 1);
        }
        return delay;
    }

    class AsyncNotifyCallBack implements Callback<String> {

        private NotifySingleTask task;

        public AsyncNotifyCallBack(NotifySingleTask task) {
            this.task = task;
        }

        @Override
        public void onReceive(RestResult<String> result) {
            if (!result.ok()) {
                //延时任务重试
                asyncTaskExecute(task);
                LOGGER.error("[notify-retry] target:{} dataId:{} group:{} ts:{}", task.target, task.getDataId(),
                        task.getGroup(), task.getLastModified());
            }
        }

        @Override
        public void onError(Throwable ex) {
            //延时任务重试
            asyncTaskExecute(task);
            LOGGER.error("[notify-retry] target:{} dataId:{} group:{} ts:{}", task.target, task.getDataId(),
                    task.getGroup(), task.getLastModified());

        }

        @Override
        public void onCancel() {
            //延时任务重试
            asyncTaskExecute(task);
            LOGGER.error("[notify-retry] target:{} dataId:{} group:{} ts:{}", task.target, task.getDataId(),
                    task.getGroup(), task.getLastModified());


        }
    }
}
